use std::{
    collections::HashMap,
    sync::{Arc, Mutex},
    time::Duration,
};

use crate::{application::Application, log};
use bytes::Bytes;
use common::{
    base::{id_generator, status::Status},
    protocol::{protocol_v1::ProtocolV1Writer, MsgType},
    serial::request_message::RequestMessage,
    service::ServiceKey,
};
use lazy_static::lazy_static;
use tokio::sync::oneshot::Sender;
use tracing::{error, warn};

use super::{local_service::LocalService, service::Service};

pub mod db_api;
pub mod file_api;
pub mod http_api;
pub mod xport_api;

lazy_static! {
    static ref SENDER_MAP: Mutex<HashMap<i64, Sender<Vec<u8>>>> = Mutex::new(HashMap::new());
    static ref REQUEST_ID_MAP: Mutex<HashMap<i64, Box<dyn Fn(Sender<Vec<u8>>, *const u8, u32) + Send + Sync>>> =
        Mutex::new(HashMap::new());
    static ref XPORT_SERVICE_KEY: ServiceKey =
        ServiceKey::new("Xport".into(), "0.0.1".into(), "Xport".into());
}

async fn get_local_service(
    service_id: i64,
    request_id: i64,
    want_system: bool,
) -> Option<Arc<Box<LocalService>>> {
    let airport = Application::get_airport();

    let dxc_manager = airport.get_dxc_manager();

    let local_service = dxc_manager.get_local_service_by_id(service_id).await;

    if local_service.is_none() {
        error!("调用的服务已经下线!");
        return None;
    }

    let local_service = local_service.unwrap();

    if want_system {
        if !local_service.is_system() {
            pack_and_return_error(
                request_id,
                local_service.get_service_key(),
                Status::error("无权限调用!".into()),
            );
            return None;
        }
    }
    Some(local_service)
}


fn pack_and_return_message<T>(request_id: i64, receiver_service_key: ServiceKey, message: T)
where
    T: RequestMessage,
{
    let status = Status::default();

    let mut message_size = status.compute_size_with_tag_and_len() as u32;

    message_size += message.compute_size_with_tag_and_len() as u32;

    let mut writer = ProtocolV1Writer::new(
        message_size,
        request_id,
        &XPORT_SERVICE_KEY,
        &receiver_service_key,
    );
    writer.write_msg_type(MsgType::RspNormalMsg);
    writer.write_conn_id(0);

    let msg_body_buffer = writer.msg_body_buffer();
    let mut os = protobuf::CodedOutputStream::bytes(msg_body_buffer);
    status.serial_with_tag_and_len(&mut os);
    message.serial_with_tag_and_len(&mut os);
    drop(os);
    tokio::spawn(async move {
        let airport = Application::get_airport();
        airport.message_in(1, writer.msg_buffer).await;
    });
}


fn pack_and_return_bool(request_id: i64, receiver_service_key: ServiceKey, value: bool) {
  let bool_size = 1 + 1;
  let mut writer = ProtocolV1Writer::new(
      bool_size,
      request_id,
      &XPORT_SERVICE_KEY,
      &receiver_service_key,
  );
  writer.write_msg_type(MsgType::RspNormalMsg);
  writer.write_conn_id(0);

  let msg_body_buffer = writer.msg_body_buffer();
  let mut os = protobuf::CodedOutputStream::bytes(msg_body_buffer);

  os.write_bool(1, value).unwrap();
  drop(os);
  tokio::spawn(async move {
      let airport = Application::get_airport();
      airport.message_in(0, writer.msg_buffer).await;
  });
}

fn pack_and_return_error(request_id: i64, receiver_service_key: ServiceKey, status: Status) {
    pack_and_return_status(request_id, receiver_service_key, status);
}

fn pack_and_return_success(request_id: i64, receiver_service_key: ServiceKey) {
    pack_and_return_status(request_id, receiver_service_key, Status::default());
}

fn pack_and_return_status(request_id: i64, receiver_service_key: ServiceKey, status: Status) {
    let status_size = status.compute_size_with_tag_and_len() as u32;
    let mut writer = ProtocolV1Writer::new(
        status_size,
        request_id,
        &XPORT_SERVICE_KEY,
        &receiver_service_key,
    );
    writer.write_msg_type(MsgType::RspNormalMsg);
    writer.write_conn_id(0);

    let msg_body_buffer = writer.msg_body_buffer();
    let mut os = protobuf::CodedOutputStream::bytes(msg_body_buffer);
    status.serial_with_tag_and_len(&mut os);
    drop(os);
    tokio::spawn(async move {
        let airport = Application::get_airport();
        airport.message_in(1, writer.msg_buffer).await;
    });
}

fn pack_and_return_empty(request_id: i64, receiver_service_key: ServiceKey) {
    let mut writer =
        ProtocolV1Writer::new(0, request_id, &XPORT_SERVICE_KEY, &receiver_service_key);
    writer.write_msg_type(MsgType::RspNormalMsg);
    writer.write_conn_id(0);

    tokio::spawn(async move {
        let airport = Application::get_airport();
        airport.message_in(1, writer.msg_buffer).await;
    });
}

fn pack_and_return_status_and_bytes(
    request_id: i64,
    receiver_service_key: ServiceKey,
    status: Status,
    bytes: Bytes,
) {
    let status_size = status.compute_size_with_tag_and_len() as u32 + bytes.len() as u32;

    let mut writer = ProtocolV1Writer::new(
        status_size,
        request_id,
        &XPORT_SERVICE_KEY,
        &receiver_service_key,
    );

    writer.write_msg_type(MsgType::RspNormalMsg);

    writer.write_conn_id(0);

    let msg_body_buffer = writer.msg_body_buffer();
    let mut os = protobuf::CodedOutputStream::bytes(msg_body_buffer);
    status.serial_with_tag_and_len(&mut os);

    let slice_bytes = bytes.as_ref();
    os.write_raw_bytes(slice_bytes).unwrap();

    drop(os);
    tokio::spawn(async move {
        let airport = Application::get_airport();
        airport.message_in(1, writer.msg_buffer).await;
    });
}

pub fn add_response_handler(
    request_id: i64,
    sender: Sender<Vec<u8>>,
    handler: Box<dyn Fn(Sender<Vec<u8>>, *const u8, u32) + Send + Sync>,
) {
    {
        let mut sender_map = SENDER_MAP.lock().unwrap();
        sender_map.insert(request_id, sender);
    }
    //
    {
        let mut handler_map = REQUEST_ID_MAP.lock().unwrap();
        handler_map.insert(request_id, handler);
    }
}

/**
*
*/
pub extern "C" fn gen_id(service_id: i64) -> i64 {
    id_generator::gen_id()
}

pub extern "C" fn sleep(service_id: i64, request_id: i64, milliseconds: u64) {
    tokio::spawn(async move {
        {
            let airport = Application::get_airport();
            let dxc_manager = airport.get_dxc_manager();

            let local_service = dxc_manager.get_local_service_by_id(service_id).await;

            if local_service.is_none() {
                error!("调用的服务已经下线!");
                return;
            }
        }

        tokio::time::sleep(Duration::from_millis(milliseconds)).await;

        let airport = Application::get_airport();

        let dxc_manager = airport.get_dxc_manager();
        let local_service = dxc_manager.get_local_service_by_id(service_id).await;
        if local_service.is_none() {
            error!("调用的服务已经下线!");
            return;
        }
        pack_and_return_success(request_id, local_service.unwrap().get_service_key());
    });
}

/**
 *
 */
pub extern "C" fn response(_service_id: i64, request_id: i64, msg: *const u8, len: u32) {
    // 回应消息
    let sender = {
        let mut sender_map = SENDER_MAP.lock().unwrap();
        sender_map.remove(&request_id)
    };

    if sender.is_none() {
        warn!("找不到请求id {} 的sender", request_id);
        return;
    }
    //
    let handler = {
        let mut handler_map = REQUEST_ID_MAP.lock().unwrap();
        handler_map.remove(&request_id)
    };

    if handler.is_none() {
        warn!("找不到请求id {} 的 handler", request_id);
        return;
    }

    let sender = sender.unwrap();
    let handler = handler.unwrap();

    handler(sender, msg, len);
}

pub extern "C" fn log_output(_service_id: i64, log: *const u8, log_len: u32) {
    let slice = unsafe { std::slice::from_raw_parts(log, log_len as usize) };

    let log_vec = slice.to_vec();
    log::append_log(log_vec);
}

//
